package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	//	"bytes"

	"bytes"
	"fmt"
	"io/ioutil"
	"labgob"
	"log"
	"math/rand"
	"sort"
	"sync"
	"sync/atomic"
	"time"

	//	"6.824/labgob"
	"labrpc"
)

const (
	kFollower   = "follower"
	kCandidiate = "candidate"
	kLeader     = "leader"

	kElectionTimeoutLower = 150
	kElectionTimeoutUpper = 300
	kHeartbeatFrequence   = 100
)

type State string

type LogEntry struct {
	Command interface{}
	Term    int
}

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	// 持久
	currentTerm int        // 当前任期
	votedFor    int        // 给谁投票了
	log         []LogEntry // 此server持有的所有日志条目

	// 易失
	commitIndex int   // 已提交的最后一个条目的索引
	lastApplied int   // 以应用的最后一个条目的索引
	state       State // 状态：跟随、候选、领导

	// leader 独有
	nextIndex  []int // 每个server下一个日志条目的索引
	matchIndex []int // 每个server可确定的持有的最后一个日志条目

	timeout       int  // 剩余超时时间
	isTimeoutStop bool // 是否停止倒计时
	leaderCond    sync.Cond
	applyCh       chan ApplyMsg
}

func (rf *Raft) String() string {
	return fmt.Sprintf("server %d info, term: %d, votedFor: %d, log: %v, commitIdx: %d, lastApplied: %d, state: %s\n",
		rf.me, rf.currentTerm, rf.votedFor, rf.log, rf.commitIndex, rf.lastApplied, rf.state)
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {

	var term int
	var isleader bool
	// Your code here (2A).
	rf.mu.Lock()
	term = rf.currentTerm
	isleader = rf.state == kLeader
	rf.mu.Unlock()

	return term, isleader
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.currentTerm)
	e.Encode(rf.votedFor)
	e.Encode(rf.log)
	data := w.Bytes()
	rf.persister.SaveRaftState(data)

}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var currentTerm int
	var voteFor int
	var entries []LogEntry
	if d.Decode(&currentTerm) != nil ||
		d.Decode(&voteFor) != nil ||
		d.Decode(&entries) != nil {
		log.Fatalf("readpersist fail")
	} else {
		rf.mu.Lock()
		rf.currentTerm = currentTerm
		rf.votedFor = voteFor
		rf.log = entries
		rf.mu.Unlock()
	}
}

//
// A service wants to switch to snapshot.  Only do so if Raft hasn't
// have more recent info since it communicate the snapshot on applyCh.
//
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {

	// Your code here (2D).

	return true
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).

}

//
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int // 候选人任期
	CandidateId  int // 候选人id
	LastLogIndex int // 候选人最后一个日志条目的下标
	LastLogTerm  int // 候选人最后一个认知条目的任期
}

func (args *RequestVoteArgs) String() string {
	return fmt.Sprintf("{term: %d, candidateId: %d}", args.Term, args.CandidateId)
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int  // 投票人的任期
	VoteGranted bool // 是否投同意票
}

func (reply *RequestVoteReply) String() string {
	return fmt.Sprintf("{term: %d, voteGranted: %t}", reply.Term, reply.VoteGranted)
}

//
// example RequestVote RPC handler.
//
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()
	defer rf.mu.Unlock()

	// 任期小于自己，直接false
	if args.Term < rf.currentTerm {
		reply.VoteGranted = false
		reply.Term = rf.currentTerm
		log.Println("request vote fail: ", args.CandidateId, "->", rf.me, "  ,caused by low term")
		return
	}

	// 任期大于自己，退为follower，更新term
	if args.Term > rf.currentTerm {
		rf.state = kFollower
		rf.currentTerm = args.Term
		rf.votedFor = -1
		rf.persist()
	}

	// 已经投过票  或  日志没自己新，不投票
	if rf.votedFor != -1 || rf.log[len(rf.log)-1].Term > args.LastLogTerm || rf.log[len(rf.log)-1].Term == args.LastLogTerm && len(rf.log)-1 > args.LastLogIndex {
		reply.VoteGranted = false
		reply.Term = rf.currentTerm

		if rf.votedFor != -1 {
			log.Println("request vote fail: ", args.CandidateId, "->", rf.me, "  ,caused by having voted")

		} else {
			log.Println("request vote fail: ", args.CandidateId, "->", rf.me, "  ,caused by out-of-date last log")
		}
		return
	}

	rf.resetElectionTimeout()
	rf.votedFor = args.CandidateId
	rf.persist()

	reply.Term = rf.currentTerm
	reply.VoteGranted = true

	log.Println("request vote success: ", args.CandidateId, "->", rf.me, "args: ", args, "reply: ", reply, "server state:", rf.state)
}

type AppendEntriesArgs struct {
	Term         int        // leader的任期
	LeaderId     int        // leader id
	PrevLogIndex int        //
	PrevLogTerm  int        //
	Entries      []LogEntry // 心跳包，此项设为空，否之设为leader的log
	LeaderCommit int        // leader的commitIndex
}

func (args *AppendEntriesArgs) String() string {
	return fmt.Sprintf("{term: %d, leaderId: %d, preIdx: %d, preTerm: %d, entries: %v, leaderCommit: %d}", args.Term, args.LeaderId, args.PrevLogIndex, args.PrevLogTerm, args.Entries, args.LeaderCommit)
}

type AppendEntriesReply struct {
	Term    int  // 此server的任期
	Success bool // 日志追加是否成功

	ConflictTerm  int // 冲突的日志的term
	ConflictIndex int // 自己存储的冲突任期的第一个idx
}

// AppendEntries RPC handler
func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	// 任期比自己小，直接返回
	if args.Term < rf.currentTerm {
		reply.Success = false
		reply.Term = rf.currentTerm
		log.Printf("append entries fail:  %d-> %d , caused by low term, args:{%v}\n", args.LeaderId, rf.me, args)
		return
	}

	reply.Term = rf.currentTerm
	rf.resetElectionTimeout()

	// 任期大于等于自己，更新任期，退为follower。
	//      可能是candidate：例如两个candidate共同竞选，一个赢了，另一个便会收到获胜者的appendEntries请求
	// 		也可能是leader：例如leader网络不畅，只是有一个follower参与选举，这个follower变成candidate后，term就有可能大于旧leader
	if args.Term > rf.currentTerm {
		rf.currentTerm = args.Term
		rf.votedFor = args.LeaderId
		rf.state = kFollower
		rf.persist()
	}

	// prev不匹配
	if len(rf.log)-1 < args.PrevLogIndex || rf.log[args.PrevLogIndex].Term != args.PrevLogTerm {
		reply.Success = false

		// conflict相关
		if len(rf.log)-1 < args.PrevLogIndex { // 因为不含有pre
			reply.ConflictIndex = len(rf.log)
			reply.ConflictTerm = -1
		} else { // 含有pre，但是term不匹配
			reply.ConflictTerm = rf.log[args.PrevLogIndex].Term
			for i := args.PrevLogIndex; i >= 0 && rf.log[i].Term == reply.ConflictTerm; i-- {
				reply.ConflictIndex = i
			}
		}

		log.Printf("append entries fail:  %d-> %d , caused by wrong prev log, args:{%v}\n", args.LeaderId, rf.me, args)
		return
	}

	// 更新log
	var i, j int
	for i, j = args.PrevLogIndex+1, 0; i < len(rf.log) && j < len(args.Entries); i, j = i+1, j+1 {
		if rf.log[i].Term != args.Entries[j].Term {
			rf.log = rf.log[:i]
			rf.persist()
			break
		}
	}
	if j != len(args.Entries) { // rf.log太短 或  有冲突
		rf.log = append(rf.log[:i], args.Entries[j:]...)
		rf.persist()
		i += len(args.Entries) - j - 1 // index of last new entry
	}

	// for debug
	oldCommitIdx := rf.commitIndex
	oldApplyIdx := rf.lastApplied

	// 更新commitIdx
	if args.LeaderCommit > rf.commitIndex {
		rf.commitIndex = min(args.LeaderCommit, i) // 不是明显leaderCommit更小吗?
		go rf.applyLog()
	}

	reply.Success = true

	log.Printf("append entries success: %d -> %d, args: %v, reply: %v, server %d log: %v, oldCommitIdx: %d, oldApplyIdx: %d, commitIdx: %d, applyIdx: %d\n",
		args.LeaderId, rf.me, args, reply, rf.me, rf.log, oldCommitIdx, oldApplyIdx, rf.commitIndex, rf.lastApplied)
}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

func (rf *Raft) callSendRequestVote(server int, args RequestVoteArgs) bool {
	reply := RequestVoteReply{}
	ok := rf.sendRequestVote(server, &args, &reply)

	rf.mu.Lock()
	defer rf.mu.Unlock()

	if !ok {
		log.Printf("%d call server %d: Raft.sendRequestVote fail\n", rf.me, server)
		return false
	}

	// 有更大的任期，退为follower
	if reply.Term > rf.currentTerm {
		rf.currentTerm = reply.Term
		rf.votedFor = -1
		rf.state = kFollower
		rf.persist()
	}

	return reply.VoteGranted
}

func (rf *Raft) callSendAppendEntries(server int, args *AppendEntriesArgs) {
	reply := AppendEntriesReply{}
	// fmt.Println("before append to", server)
	ok := rf.sendAppendEntries(server, args, &reply)
	if !ok {
		log.Printf("%d call server %d: Raft.AppendEntries fail\n", args.LeaderId, server)
		return
	}

	// 处理reply
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if reply.Term > rf.currentTerm { // 因为自己任期过时
		rf.state = kFollower
		rf.currentTerm = reply.Term
		rf.votedFor = -1
		rf.persist()
		return
	}

	if reply.Success { // append成功
		rf.matchIndex[server] = args.PrevLogIndex + len(args.Entries)
		rf.nextIndex[server] = rf.matchIndex[server] + 1
	} else { // 因为prev匹配失败
		rf.nextIndex[server] = reply.ConflictIndex

		if reply.ConflictTerm != -1 {
			for i := args.PrevLogIndex; i >= 1; i-- {
				if rf.log[i-1].Term == reply.ConflictTerm {
					rf.nextIndex[server] = i
					break
				}
			}
		}
	}
	log.Println("		after append to", server, ", next:", rf.nextIndex, "match:", rf.matchIndex)
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := true

	// Your code here (2B).
	rf.mu.Lock()
	defer rf.mu.Unlock()

	index = len(rf.log)
	term = rf.currentTerm
	isLeader = rf.state == kLeader

	if isLeader {
		rf.log = append(rf.log, LogEntry{Term: rf.currentTerm, Command: command})
		rf.persist()
		rf.matchIndex[rf.me] = len(rf.log) - 1
		log.Printf("%d receive new command %v,commitIdx: %d, applyIdx: %d, current log: %v\n", rf.me, command, rf.commitIndex, rf.lastApplied, rf.log)
	}

	return index, term, isLeader
}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

// The ticker go routine starts a new election if this peer hasn't received
// heartsbeats recently.
func (rf *Raft) ticker() {
	rf.mu.Lock()
	rf.resetElectionTimeout() // 选举超时时间设初值
	rf.mu.Unlock()

	for rf.killed() == false {

		// Your code here to check if a leader election should
		// be started and to randomize sleeping time using
		// time.Sleep().
		time.Sleep(time.Millisecond)

		rf.mu.Lock()
		if rf.isTimeoutStop == false {
			rf.timeout--
		}
		if rf.timeout == 0 {
			rf.resetElectionTimeout()
			rf.mu.Unlock()
			go rf.attemptElection()
		} else {
			rf.mu.Unlock()
		}
	}
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	log.Print("Make start!!!!!!!!!!!!!!!!!!!!!!")
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me

	// Your initialization code here (2A, 2B, 2C).

	log.SetOutput(ioutil.Discard) // 禁用log
	rf.commitIndex = 0
	rf.lastApplied = 0
	rf.state = kFollower
	rf.isTimeoutStop = false
	rf.leaderCond = sync.Cond{L: &rf.mu}
	rf.applyCh = applyCh

	rf.votedFor = -1
	rf.log = make([]LogEntry, 1) // 第一个元素为哨兵
	rf.log[0].Term = 0
	rf.currentTerm = 0

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	// fmt.Printf("read persist\n")

	// start ticker goroutine to start elections
	go rf.ticker()

	go rf.sendHeartbeat(len(peers))
	go rf.updateCommit(len(peers))

	return rf
}

func (rf *Raft) attemptElection() {
	rf.mu.Lock()

	rf.state = kCandidiate
	rf.currentTerm++
	rf.votedFor = rf.me
	rf.persist()

	log.Println(rf.me, " attempt election, term = ", rf.currentTerm)

	args := RequestVoteArgs{
		Term:         rf.currentTerm,
		CandidateId:  rf.me,
		LastLogIndex: len(rf.log) - 1,
		LastLogTerm:  rf.log[len(rf.log)-1].Term,
	}
	peersNum := len(rf.peers)

	rf.mu.Unlock()

	var mutex sync.Mutex
	cond := sync.Cond{L: &mutex}
	win := 1
	finished := 1
	for i := range rf.peers {
		rf.mu.Lock()
		if i == rf.me {
			rf.mu.Unlock()
			continue
		}
		rf.mu.Unlock()

		go func(server int) {
			voteGranted := rf.callSendRequestVote(server, args)
			mutex.Lock()
			defer mutex.Unlock()
			if voteGranted {
				win++
			}
			finished++
			cond.Broadcast()
		}(i)
	}

	mutex.Lock()
	for win <= peersNum/2 && finished != peersNum {
		cond.Wait()
	}
	isWin := win > peersNum/2
	mutex.Unlock()

	rf.mu.Lock()
	defer rf.mu.Unlock()
	if isWin && rf.state == kCandidiate {
		// 赢得选举
		rf.state = kLeader
		rf.nextIndex = make([]int, peersNum)
		rf.matchIndex = make([]int, peersNum)
		for i := range rf.nextIndex {
			if i == rf.me {
				rf.matchIndex[i] = len(rf.log) - 1
			} else {
				rf.nextIndex[i] = len(rf.log)
				rf.matchIndex[i] = 0
			}
		}

		rf.leaderCond.Broadcast() // leader专属线程开始干活
		rf.isTimeoutStop = true

		log.Printf("**********  %d become leader, term: %d  **************  %s\n", rf.me, rf.currentTerm, rf.String())
	}
}

func (rf *Raft) resetElectionTimeout() {
	rf.timeout = rand.Intn(kElectionTimeoutUpper-kElectionTimeoutLower) + kElectionTimeoutLower
	rf.isTimeoutStop = false
	// log.Print(rf.me, " reset election timeout, ", rf.timeout)
}

// 开一个线程运行
// 发送心跳包
// 心跳包并不是一个特别的appendEntries包，就是一个普通的applendEntries
func (rf *Raft) sendHeartbeat(peersNum int) {
	for rf.killed() == false {
		rf.mu.Lock()
		for rf.state != kLeader {
			rf.leaderCond.Wait()
		}
		rf.mu.Unlock()

		for i := 0; i < peersNum; i++ {
			rf.mu.Lock() ////
			if i == rf.me {
				rf.mu.Unlock()
				continue
			}

			// 准备参数
			prevLogIdx := rf.nextIndex[i] - 1
			entries := append([]LogEntry{}, rf.log[prevLogIdx+1:len(rf.log)]...)
			args := AppendEntriesArgs{
				Term:         rf.currentTerm,
				LeaderId:     rf.me,
				PrevLogIndex: prevLogIdx,
				PrevLogTerm:  rf.log[prevLogIdx].Term,
				Entries:      entries,
				LeaderCommit: rf.commitIndex,
			}
			rf.mu.Unlock() ////

			go rf.callSendAppendEntries(i, &args)
		}

		time.Sleep(kHeartbeatFrequence * time.Millisecond)
	}
}

// 开一个线程运行
// leader专用
// 检测并更新leader的已提交
func (rf *Raft) updateCommit(peersNum int) {
	for rf.killed() == false {
		rf.mu.Lock()
		for rf.state != kLeader {
			rf.leaderCond.Wait() // 等待成为leader
		}

		matchIdxs := append([]int{}, rf.matchIndex...)
		sort.Sort(sort.Reverse(sort.IntSlice(matchIdxs))) // 降序排序
		// fmt.Println("sorted matchIdxs:", matchIdxs)
		majorMatchIdx := matchIdxs[peersNum/2] // 大多数的已拷贝的条目下标

		if majorMatchIdx > rf.commitIndex && rf.log[majorMatchIdx].Term == rf.currentTerm {
			rf.commitIndex = majorMatchIdx // 更新已提交
			log.Printf("leader %d update commitIdx, commitIdx = %d, applyIdx = %d\n", rf.me, rf.commitIndex, rf.lastApplied)
			go rf.applyLog()
		}

		rf.mu.Unlock()

		time.Sleep(50 * time.Millisecond)
	}
}

// 当更新commitIdx时，开一个线程执行此函数，apply日志
// 所有server都能用
// 检测commit>apply，如果大于，则逐条apply log
func (rf *Raft) applyLog() {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if rf.lastApplied < rf.commitIndex {
		for i := rf.lastApplied + 1; i <= rf.commitIndex; i++ {
			applyMsg := ApplyMsg{
				CommandValid: true,
				Command:      rf.log[i].Command,
				CommandIndex: i,
			}
			rf.applyCh <- applyMsg
			rf.lastApplied++
			// log.Println(rf.me, "apply one log, after apply, commitIdx:", rf.commitIndex, "applyIdx:", rf.lastApplied)
			// fmt.Println("apply msg", applyMsg)
		}
		log.Println(rf.me, "apply some logs, after apply, commitIdx:", rf.commitIndex, "applyIdx:", rf.lastApplied)
	}

}

func min(x, y int) int {
	if x < y {
		return x
	} else {
		return y
	}
}
